package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/2/6
 *
 *  每隔5分钟输出最近1小时内点击量最多的前N(3)个商品
 */
public class Demo8_TopN
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        //创建表映射数据
        String createTableSql = "create table t1( uid bigint,itemId bigint ,cid int , behavior string, ts bigint ," +
            "                      et as  TO_TIMESTAMP_LTZ(ts, 0)  ," +
            "                    WATERMARK FOR et AS et - INTERVAL '0.001' SECOND  " +
            " ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/UserBehavior.csv' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        //执行建表
        tableEnvironment.executeSql(createTableSql);

        //第一次聚合: 开窗，统计每种商品的点击次数
        String hop = "SELECT window_start, window_end, itemId , count(*) click" +
            "  FROM TABLE(" +
            "    HOP(TABLE t1, DESCRIPTOR(et), INTERVAL '5' MINUTES ,INTERVAL '1' HOURS))" +
            "  GROUP BY window_start, window_end , itemId" ;

        Table table = tableEnvironment.sqlQuery(hop);
        tableEnvironment.createTemporaryView("t2",table);

        /*
            第二次聚合:  求每个窗口中的TopN
                先排名，再过滤
                排名函数:
                    hive : rank, dense_rank,row_number
                    flink: row_number
                 开窗函数。  row_number over()
         */
        String topN = "select window_start, window_end, itemId, click ," +
            "                 row_number() over(partition by  window_start order by click desc ) rk " +
            "          from t2";

        Table table1 = tableEnvironment.sqlQuery(topN);
        tableEnvironment.createTemporaryView("t3",table1);

        /*
            The window can only be ordered in ASCENDING mode.
             OVER windows' ordering in stream mode must be defined on a time attribute

             over()window中order by后续只能更时间字段，如果是非时间字段，必须和过滤一起使用！
         */
        //table1.execute().print();

        String outputTableSql = " CREATE TABLE `t4` (" +
            " `w_start` TIMESTAMP  ," +
            "  `w_end` TIMESTAMP ," +
            "  `item_id` BIGINT," +
            "  `item_count` BIGINT," +
            "  `rk` BIGINT," +
            "  PRIMARY KEY (`w_end`,`rk`) NOT ENFORCED " +
            "     )WITH ( " +
            "     'username' = 'root' ," +
            "      'password' = '000000' , " +
            "        'table-name' = 'hot_item' ," +
            "    'connector' = 'jdbc',  " +
            "    'url' = 'jdbc:mysql://hadoop104:3306/220828?useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true' " +
            ") ";

        tableEnvironment.executeSql(outputTableSql);
        //求前top3,把结果输出到数据库中。
        tableEnvironment.executeSql("insert into t4 select window_start, window_end, itemId, click , rk from t3 where rk <= 3");


    }
}
